package net.flyinggroup.introduction;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static net.flyinggroup.introduction.Constants.DATA_LENGTH;
import static net.flyinggroup.introduction.Constants.TCP_PORT;

public class FireNForgetClient {
    private final RSocket socket;

    public final List<Float> data = Collections.unmodifiableList(generateData());

    public FireNForgetClient() {
        this.socket = RSocketConnector.connectWith(TcpClientTransport.create(TCP_PORT)).block();
    }

    public void sendData() {
        Flux.interval(Duration.ofMillis(50))
                .take(data.size())
                .map(this::createFloatPayload)
                .flatMap(socket::fireAndForget)
                .blockLast();
    }

    private Payload createFloatPayload(Long index) {
        float velocity = data.get(index.intValue());
        ByteBuffer buffer = ByteBuffer.allocate(4).putFloat(velocity);
        buffer.rewind();
        return DefaultPayload.create(buffer);
    }

    private List<Float> generateData() {
        List<Float> dataList = new ArrayList<>(DATA_LENGTH);
        float velocity = 0;
        for (int i = 0; i < DATA_LENGTH; i++) {
            velocity += Math.random();
            dataList.add(velocity);
        }
        return dataList;
    }

    public List<Float> getData() {
        return data;
    }

    public void dispose() {
        socket.dispose();
    }
}
